package com.lzt.app

import java.util.Properties
import javax.servlet.{ServletContextEvent, ServletContextListener}

import org.json4s._
import org.json4s.jackson.JsonMethods._
import org.scalatra.atmosphere._
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.scalatra.atmosphere.AtmosphereClient
import org.slf4j.LoggerFactory

import scala.collection.JavaConverters
import scala.concurrent.ExecutionContext.Implicits.global

class KafkaListener extends ServletContextListener {

  implicit protected val jsonFormats: Formats = DefaultFormats
  val logger = LoggerFactory.getLogger(classOf[KafkaListener])
  var contextUrl = ""

  lazy val thread = new Thread(() => {
    val props = new Properties()
    props.put("bootstrap.servers", Config.kafkaServers)
    props.put("group.id", "test")
    props.put("enable.auto.commit", "false")
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val consumer = new KafkaConsumer[String, String](props)

    consumer.subscribe(JavaConverters.asJavaCollection(List(Config.kafkaNoticeTopic)))

    while (true) {
      val records = consumer.poll(Long.MaxValue)
      records.forEach(x => {
        logger.debug(s"recv kafka message : key = ${x.key()}, value = ${x.value()}")
        //目前接收到的都是 tccgl 服务的消息，暂不需要考虑订阅的情况
        try {
//          val data = parse(x.value())
          AtmosphereClient.broadcast(s"$contextUrl/socket/ws", TextMessage(x.value()), SjtpAtmosphereClient.tccglFilter)
        } catch {
          case e: Exception => logger.error("解析 kafka message 异常：", e)
        }
      })
    }
  })

  override def contextInitialized(sce: ServletContextEvent): Unit = {
    contextUrl = sce.getServletContext.getContextPath
    thread.start()
  }

  override def contextDestroyed(sce: ServletContextEvent): Unit = {
    thread.interrupt()
  }
}
